package com.zhss.dfs.server;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;

/**
 * @program: filesystem
 * @description:
 * @author: Liu Weidong
 * @create: 2024-05-08 22:55
 **/
public class FsImageServer extends Thread {

    private Selector selector;

    private String fsImage = "D://dfs//image.meta";

    public FsImageServer() throws IOException {
        initServer(60006);
    }


    @Override
    public void run() {
        try {
            listen();
        } catch (IOException ioException) {
            ioException.printStackTrace();
        }
    }

    /**
     * 进行初始化工作
     *
     * @param port
     * @throws IOException
     */
    public void initServer(int port) throws IOException {
        // 打开serverSocket通道
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        // 设置为非阻塞模式
        serverSocketChannel.configureBlocking(false);
        // 绑定相关端口
        serverSocketChannel.socket().bind(new InetSocketAddress(port));
        // 打开一个selector
        this.selector = Selector.open();
        // 将channel与selector进行绑定，并注册相关事件
        serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT);
    }

    public void listen() throws IOException {
        System.out.println("nameNode的fsImage服务器启动，监听端口为60006.......");
        while (true) {
            selector.select();
            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                iterator.remove();
                handler(key);
            }
        }
    }

    private void handler(SelectionKey key) throws IOException {
        if (key.isAcceptable()) {
            // 接收客户端连接请求
            handlerAccept(key);
        } else if (key.isReadable()) {
            // 处理客户端读请求
            handlerRead(key);
        } else if (key.isWritable()) {
            // 处理客户端写请求
            handlerWrite(key);
        }
    }

    private void handlerWrite(SelectionKey key) throws IOException {
        SocketChannel socketChannel = null;
        try {
            socketChannel = (SocketChannel) key.channel();
            ByteBuffer wrap = ByteBuffer.wrap("SUCCESS".getBytes(StandardCharsets.UTF_8));
            socketChannel.write(wrap);
            socketChannel.register(selector, SelectionKey.OP_READ);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (socketChannel != null) {
                socketChannel.close();
            }
        }
    }

    private void handlerRead(SelectionKey key) throws IOException {
        SocketChannel socketChannel = (SocketChannel) key.channel();
        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
        long readCount = 0;
        // 先删除上一个fsImage文件，再写入新的
        File file = new File(fsImage);
        if (file.exists()) {
            file.delete();
        }
        FileOutputStream fos = null;
        FileChannel channel = null;
        try {
            fos = new FileOutputStream(fsImage);
            channel = fos.getChannel();
            while ((readCount = socketChannel.read(byteBuffer)) > 0) {
                byteBuffer.flip();
                channel.write(byteBuffer);
                byteBuffer.clear();
            }
            System.out.println("namenode:fsImage刷盘成功....");
            socketChannel.register(selector, SelectionKey.OP_WRITE);
        } catch (Exception e) {
            e.printStackTrace();
            if (socketChannel != null) {
                socketChannel.close();
            }
        } finally {
            if (fos != null) {
                fos.close();
            }
            if (channel != null) {
                channel.close();
            }
        }

    }

    private void handlerAccept(SelectionKey key) throws IOException {
        ServerSocketChannel server = (ServerSocketChannel) key.channel();
        SocketChannel socketChannel = server.accept();
        socketChannel.configureBlocking(false);
        // 注册读事件
        socketChannel.register(selector, SelectionKey.OP_READ);
    }

}
